{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "707ea67a-7780-4277-8040-52dd39202ca1",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.functions as F\n",
    "import pyspark.sql.types as T\n",
    "\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "362d655c-8c79-4ed1-aee2-c1707f215917",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"counting word occurences from a book,one more time.\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "0617803a-617a-4c06-9dff-88c4ddc120fc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result = (\n",
    "    spark.read.text(r\"D:\\bigdata\\spark_ practice\\DataAnalysisWithPythonAndPySpark-Data-trunk\\gutenberg_books\\2701-0.txt\")\n",
    "    .select(F.split(F.col(\"value\"),\" \").alias(\"line\"))\n",
    "    .select(F.explode(F.col(\"line\")).alias(\"word\"))\n",
    "    .select(F.lower(F.col(\"word\")).alias(\"word\"))\n",
    "    .select(F.regexp_extract(F.col(\"word\"),\"[a-z]+\",0).alias(\"word\"))\n",
    "    .where(F.col(\"word\") != \"\")\n",
    "    .groupBy(F.col(\"word\"))\n",
    "    .count()\n",
    ")\n",
    "result.orderBy(\"count\",ascending=False).show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "e28f19ed-dc17-4bf8-9f96-7cc6f7e6a08b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Parsed Logical Plan ==\n",
      "'Aggregate ['word], ['word, count(1) AS count#14L]\n",
      "+- Filter NOT (word#10 = )\n",
      "   +- Project [regexp_extract(word#8, [a-z]+, 0) AS word#10]\n",
      "      +- Project [lower(word#6) AS word#8]\n",
      "         +- Project [word#6]\n",
      "            +- Generate explode(line#2), false, [word#6]\n",
      "               +- Project [split(value#0,  , -1) AS line#2]\n",
      "                  +- Relation [value#0] text\n",
      "\n",
      "== Analyzed Logical Plan ==\n",
      "word: string, count: bigint\n",
      "Aggregate [word#10], [word#10, count(1) AS count#14L]\n",
      "+- Filter NOT (word#10 = )\n",
      "   +- Project [regexp_extract(word#8, [a-z]+, 0) AS word#10]\n",
      "      +- Project [lower(word#6) AS word#8]\n",
      "         +- Project [word#6]\n",
      "            +- Generate explode(line#2), false, [word#6]\n",
      "               +- Project [split(value#0,  , -1) AS line#2]\n",
      "                  +- Relation [value#0] text\n",
      "\n",
      "== Optimized Logical Plan ==\n",
      "Aggregate [word#10], [word#10, count(1) AS count#14L]\n",
      "+- Project [regexp_extract(lower(word#6), [a-z]+, 0) AS word#10]\n",
      "   +- Filter NOT (regexp_extract(lower(word#6), [a-z]+, 0) = )\n",
      "      +- Generate explode(line#2), [0], false, [word#6]\n",
      "         +- Project [split(value#0,  , -1) AS line#2]\n",
      "            +- Filter ((size(split(value#0,  , -1), true) > 0) AND isnotnull(split(value#0,  , -1)))\n",
      "               +- Relation [value#0] text\n",
      "\n",
      "== Physical Plan ==\n",
      "AdaptiveSparkPlan isFinalPlan=false\n",
      "+- HashAggregate(keys=[word#10], functions=[count(1)], output=[word#10, count#14L])\n",
      "   +- Exchange hashpartitioning(word#10, 200), ENSURE_REQUIREMENTS, [plan_id=113]\n",
      "      +- HashAggregate(keys=[word#10], functions=[partial_count(1)], output=[word#10, count#24L])\n",
      "         +- Project [regexp_extract(lower(word#6), [a-z]+, 0) AS word#10]\n",
      "            +- Filter NOT (regexp_extract(lower(word#6), [a-z]+, 0) = )\n",
      "               +- Generate explode(line#2), false, [word#6]\n",
      "                  +- Project [split(value#0,  , -1) AS line#2]\n",
      "                     +- Filter ((size(split(value#0,  , -1), true) > 0) AND isnotnull(split(value#0,  , -1)))\n",
      "                        +- FileScan text [value#0] Batched: false, DataFilters: [(size(split(value#0,  , -1), true) > 0), isnotnull(split(value#0,  , -1))], Format: Text, Location: InMemoryFileIndex(1 paths)[file:/D:/bigdata/spark_ practice/DataAnalysisWithPythonAndPySpark-Data..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<value:string>\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result.explain(extended=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "7c914b2f-1b9b-4551-a5a6-0736b7917c1a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "507 ms ± 31.3 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)\n"
     ]
    }
   ],
   "source": [
    "%timeit result.orderBy(\"count\",ascending=False).show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "cd39911f-e645-4161-a311-f1b3f607a9a0",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "df_init = (\n",
    "    spark.read.text(r\"D:\\bigdata\\spark_ practice\\DataAnalysisWithPythonAndPySpark-Data-trunk\\gutenberg_books\\2701-0.txt\")\n",
    "    .select(F.split(F.col(\"value\"),\" \").alias(\"line\"))\n",
    "    .select(F.explode(F.col(\"line\")).alias(\"word\"))\n",
    "    .select(F.lower(F.col(\"word\")).alias(\"word\"))\n",
    "    .select(F.regexp_extract(F.col(\"word\"),\"[a-z]+\",0).alias(\"word\"))\n",
    "    .cache()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "7fc7bf8c-70aa-41b9-bcc3-2d16cf6120f3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "| the|14603|\n",
      "|  of| 6714|\n",
      "| and| 6451|\n",
      "|   a| 4733|\n",
      "|  to| 4684|\n",
      "|  in| 4216|\n",
      "|that| 3044|\n",
      "| his| 2523|\n",
      "|  it| 2508|\n",
      "|   i| 2081|\n",
      "+----+-----+\n",
      "only showing top 10 rows\n",
      "\n",
      "317 ms ± 18.7 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)\n"
     ]
    }
   ],
   "source": [
    "%timeit df_init.where(F.col(\"word\") != \"\").groupBy(F.col(\"word\")).count().orderBy(\"count\",ascending=False).show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4b2804b9-cec8-4db7-a173-fcc999939af2",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
